package com.trz.cqrs.core.async;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.trz.cqrs.core.entity.EventEntity;
import com.trz.cqrs.core.enums.EventState;
import com.trz.cqrs.core.priority.PriorityStrategy;
import com.trz.cqrs.es.event.Event;
import com.trz.cqrs.es.event.EventHandler;
import com.trz.cqrs.es.event.EventProcessor;
import com.trz.cqrs.es.event.EventView;
import com.trz.cqrs.es.event.priority.EventPriority;
import com.trz.cqrs.es.event.priority.Priority;
import com.trz.cqrs.es.event.priority.Strategy;
import org.springframework.beans.BeanUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.transaction.support.TransactionSynchronization;

import com.trz.cqrs.core.exception.UnsupportedEventHandlerException;
import com.trz.cqrs.core.repository.EventRepository;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;


import javax.annotation.Resource;
import java.lang.reflect.ParameterizedType;
import java.time.Instant;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;

import static com.trz.cqrs.core.configurer.Constants.EXECUTOR;
import static com.trz.cqrs.core.configurer.Constants.MODULE;

import static java.time.Duration.between;
import static java.time.Instant.now;
import static java.util.Optional.ofNullable;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toMap;
import static org.apache.commons.lang3.StringUtils.substringAfterLast;
import static org.springframework.transaction.support.TransactionSynchronizationManager.isSynchronizationActive;
import static org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization;

@Slf4j
@Component
class AsyncProcessor implements EventProcessor {

    @Resource
    private RetryProcessor retryer;

    @Resource
    private EventRepository repository;

    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private ApplicationEventPublisher publisher;

    private final Map<Class<? extends Event>, EventHandler<? extends Event>> handlers;

    private final Map<Strategy, PriorityStrategy> strategies;

    AsyncProcessor(List<EventHandler<? extends Event>> handlerList,
                   List<PriorityStrategy> strategies) {

        this.handlers = handlerList.stream().collect(toMap(h -> (Class<? extends Event>)
                Stream.of(h.getClass().getGenericInterfaces())
                        .filter(x -> x instanceof ParameterizedType)
                        .map(x -> (ParameterizedType) x)
                        .findFirst()
                        .orElseGet(() -> (ParameterizedType) h.getClass().getGenericSuperclass())//h.getClass()
                        .getActualTypeArguments()[0], identity()));
        log.info("Registered {} event handler(s) in {}", this.handlers.size(), MODULE);
        this.strategies = strategies.stream().collect(toMap(PriorityStrategy::getStrategy, identity()));
        log.info("Registered {} priority strategies in {}", this.strategies.size(), MODULE);
    }

    @Override
    public void trigger(EventView view) {
        publisher.publishEvent(view);
    }

    @Override
    public void trigger(Event event) {
        persist(event);
        if (isSynchronizationActive()) {
            registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    /* trigger after transaction commit */
                    trigger(event::getId);
                }
            });
        } else {
            /* trigger immediately if without transaction */
            trigger(event::getId);
        }
    }

    @Async(EXECUTOR)
    @EventListener
    public void onEventView(EventView view) {
        ofNullable(retryer.lock(view)).ifPresent(event -> {
            final Instant start = now();
            try {
                final EventHandler<Event> handler = (EventHandler<Event>) handlers.get(event.getClass());
                ofNullable(handler)
                        .orElseThrow(() -> new UnsupportedEventHandlerException(event))
                        .handle(event, this);
                update(event);
            } catch (final UnsupportedOperationException e) {
                update(event, EventState.UNSUPPORTED, e);
            } catch (final InterruptedException e) {
                update(event, EventState.INTERRUPTED, e.getMessage());
            } catch (final Exception e) {
                update(event, EventState.FAILED, e);
            } finally {
                log.info("Processed {} [Id: {}, Count: {}, Category: {}, Identifier: {}] in {}ms",
                        substringAfterLast(event.getEventType(), "."), event.getId(),
                        event.getCount(), event.getCategory(), event.getIdentifier(), between(start, now()).toMillis());
            }
        });
    }

    private void persist(Event event) {
        if (event.getId() == null) {
            try {
                final EventEntity entity = new EventEntity();
                BeanUtils.copyProperties(event, entity);

                final EventPriority annotation = AnnotationUtils.findAnnotation(event.getClass(), EventPriority.class);
                entity.setPriority(ofNullable(annotation).map(EventPriority::value).orElse(Priority.P5))
                        .setPayload(objectMapper.writeValueAsString(event))
                        .setCreateTime(Calendar.getInstance().getTime())
                        .setUpdateTime(Calendar.getInstance().getTime())
                .setCreateUser("SYS")
                .setUpdateUser("SYS");

                repository.save(entity);
                event.setId(entity.getId());
            } catch (Exception e) {
                System.out.println(e);
                log.error("Error caught in persisting event", e);
            }
        } else {
            event.setRetry(true);
        }
    }

    private void update(Event event) {
        update(event, EventState.SUCCEEDED, null, null);
    }

    private void update(Event event, EventState eventState, Throwable cause) {
        update(event, eventState, cause, null);
    }

    private void update(Event event, EventState eventState, String message) {
        update(event, eventState, null, message);
    }

    private void update(Event event, EventState eventState, Throwable cause, String message) {
        repository.findByIdAndEventStateNot(event.getId(), EventState.SUCCEEDED).ifPresent(e -> {
            /* reset state & error */
            e.setEventState(EventState.CREATED).clearError();
            /* do nothing if success with retry */
            if (!(event.isRetry() && eventState.isSucceeded())) {
                e.setEventState(eventState);
            }
            /* fill in error if needed */
            e.onError(cause);
            /* fill in message if needed */
            ofNullable(message).ifPresent(e::shortenMessage);
            /* always adjust event priority if needed */
            val annotation = AnnotationUtils.findAnnotation(event.getClass(), EventPriority.class);
            val strategy = ofNullable(annotation).map(EventPriority::strategy).orElse(Strategy.GRADIENT_DESCENT);
            ofNullable(strategies.get(strategy)).ifPresent(s -> s.adjust(e));
            /* persist updated event */
            repository.saveOrUpdate(e.incrementAndGet());
        });
    }

}
